package fun.tan90.easy.binlog.manager.service.impl;

import fun.tan90.easy.binlog.manager.model.BinlogWorkInstance;
import fun.tan90.easy.binlog.manager.server.EbManager;
import fun.tan90.easy.binlog.manager.service.BinlogWorkerService;
import lombok.extern.slf4j.Slf4j;
import org.noear.socketd.transport.core.Reply;
import org.noear.socketd.transport.core.Session;
import org.noear.socketd.transport.core.entity.StringEntity;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/12/20 14:17
 */
@Slf4j
@Service
public class BinlogWorkerServiceImpl implements BinlogWorkerService {

    @Resource
    EbManager ebManager;

    @Override
    public List<BinlogWorkInstance> onlineWorkers() {
        List<BinlogWorkInstance> binlogWorkInstances = new ArrayList<>();
        Map<String, Session> workers = ebManager.workers();
        Map<String, Set<String>> workerHaveMasterCodeMap = ebManager.workerRunningMasterCode();
        for (Map.Entry<String, Session> entry : workers.entrySet()) {
            String key = entry.getKey();
            String[] split = key.split(":");
            Session session = entry.getValue();
            Set<String> haveMasterCodeSet = workerHaveMasterCodeMap.getOrDefault(key, new HashSet<>());
            String connectedTime = session.attrOrDefault(EbManager.SESSION_ATTR_KEY_WORKER_CONNECTED_TIME, "-");
            binlogWorkInstances.add(BinlogWorkInstance.builder()
                    .workId(key)
                    .masterCodes(haveMasterCodeSet)
                    .slaveNum(haveMasterCodeSet.size())
                    .hostname(split[0])
                    .port(split[1])
                    .connectedTime(connectedTime)
                    .workerSystemResource(ebManager.workerSystemResource(key))
                    .build());
        }
        return binlogWorkInstances;
    }

    @Override
    public void broadcastSend(String event, String content, Map<String, String> metaMap) {
        metaMap = Objects.isNull(metaMap) ? new HashMap<>() : metaMap;
        Map<String, Session> workers = ebManager.workers();
        for (Session session : workers.values()) {
            try {
                session.send(event, new StringEntity(content).metaMapPut(metaMap));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    @Override
    public Map<String, String> broadcastSendAndRequest(String event, String content, Map<String, String> metaMap) {
        Map<String, Session> workers = ebManager.workers();
        Map<String, String> resMap = new HashMap<>(workers.size());
        for (Map.Entry<String, Session> entry : workers.entrySet()) {
            Session session = entry.getValue();
            try {
                Reply reply = session.sendAndRequest(event, new StringEntity(content).metaMapPut(metaMap), 15000).await();
                resMap.put(entry.getKey(), reply.dataAsString());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        return resMap;
    }
}
